package io.github.wppli.domain.recommond.rservice;

import io.github.wppli.domain.recommond.model.entity.ProductEntity;
import io.github.wppli.domain.recommond.model.entity.UserBehaviorEntity;
import io.github.wppli.domain.recommond.repository.IProductRepository;
import io.github.wppli.domain.recommond.repository.IUserBehaviorRepository;
import lombok.AllArgsConstructor;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.springframework.stereotype.Service;
import scala.Tuple2;
import io.github.wppli.domain.recommond.compare.*;

import java.util.*;
import java.util.stream.Collectors;

@Service
@AllArgsConstructor
public class SparkRecommendationService  {

    private final IUserBehaviorRepository userBehaviorRepository;

    private final IProductRepository productRepository;

    private final JavaSparkContext javaSparkContext;

    public List<ProductEntity> getRecommendations(Long userId) {
        return getRecommendations(userId, 3);
    }

    public List<ProductEntity> getRecommendations(Long userId, int topN) {
        // 获取目标用户的行为数据
        List<UserBehaviorEntity> userBehaviors = userBehaviorRepository.queryByUserId(userId);
        if (userBehaviors.isEmpty()) {
            // 如果目标用户没有行为数据，返回热门商品
            return getTopHotProducts(topN);
        }

        // 分批获取所有用户的行为数据
        List<UserBehaviorEntity> allBehaviors = new ArrayList<>();
        int batchSize = 10000; // 每批处理1000条记录
        int offset = 0;
        while (true) {
            List<UserBehaviorEntity> batch = userBehaviorRepository.findAllWithPagination(batchSize, offset);
            allBehaviors.addAll(batch);
            offset += batchSize;
            if (batch.size() < batchSize) {
                break;
            }
        }

        // 使用 Spark 计算用户相似度
        Map<Long, Double> userSimilarities = calculateUserSimilaritiesWithSpark(userId, allBehaviors);
        if (userSimilarities.isEmpty()) {
            // 如果没有相似用户，返回热门商品
            return getTopHotProducts(topN);
        }

        // 根据相似用户的行为生成推荐
        return generateRecommendationsWithSpark(userId, userSimilarities, topN);
    }

    public List<ProductEntity> getTopHotProducts(int topN) {
        return productRepository.getTopHotProducts(topN);
    }

    public List<ProductEntity> getTopHotProducts() {
        return productRepository.getTopHotProducts(10);
    }

    /**
     * 使用 Spark 计算用户相似度
     * @param targetUserId 目标用户ID
     * @param allBehaviors 所有用户行为数据
     * @return 用户相似度列表
     */
    private Map<Long, Double> calculateUserSimilaritiesWithSpark(Long targetUserId, List<UserBehaviorEntity> allBehaviors) {

        // 将行为数据转换为 RDD
        JavaRDD<UserBehaviorEntity> userBehaviorRDD = javaSparkContext.parallelize(allBehaviors);

        // 将行为数据按用户分组
        JavaPairRDD<Long, Iterable<UserBehaviorEntity>> userBehaviorsGrouped = userBehaviorRDD.groupBy(UserBehaviorEntity::getUserId);

        // 获取目标用户的行为
        List<UserBehaviorEntity> targetUserBehaviors = userBehaviorRepository.queryByUserId(targetUserId);
        Set<Long> targetUserItems = targetUserBehaviors.stream()
                .map(UserBehaviorEntity::getProductId)
                .collect(Collectors.toSet());

        // 获取所有其他用户的行为数据
        JavaPairRDD<Long, Iterable<UserBehaviorEntity>> otherUsersBehaviors = userBehaviorsGrouped.filter(pair -> !pair._1().equals(targetUserId));
        if (otherUsersBehaviors.isEmpty()) {
            return Collections.emptyMap();
        }
        // 计算每个其他用户与目标用户的相似度
        JavaPairRDD<Long, Double> userSimilarities = otherUsersBehaviors.mapToPair(pair -> {
            Long otherUserId = pair._1();
            Iterable<UserBehaviorEntity> otherUserBehaviorsIterable = pair._2();

            Set<Long> otherUserItems = new HashSet<>();
            for (UserBehaviorEntity behavior : otherUserBehaviorsIterable) {
                otherUserItems.add(behavior.getProductId());
            }

            // 计算交集大小
            Set<Long> intersection = new HashSet<>(targetUserItems);
            intersection.retainAll(otherUserItems);
            double intersectionSize = intersection.size();

            // 计算相似度（Jaccard 系数）
            double unionSize = targetUserItems.size() + otherUserItems.size() - intersectionSize;
            if (unionSize == 0) {
                return new Tuple2<>(otherUserId, 0.0);
            } else {
                return new Tuple2<>(otherUserId, intersectionSize / unionSize);
            }
        });
        // 收集结果并转换为 Map
        return userSimilarities.collectAsMap();
    }

    /**
     * 使用 Spark 生成推荐商品
     * @param targetUserId 目标用户ID
     * @param userSimilarities 用户相似度
     * @param topN 推荐数量
     * @return 推荐商品列表
     */
    private List<ProductEntity> generateRecommendationsWithSpark(Long targetUserId, Map<Long, Double> userSimilarities, int topN) {

        // 获取目标用户已经交互过的商品
        Set<Long> targetUserItems = userBehaviorRepository.queryByUserId(targetUserId).stream()
                .map(UserBehaviorEntity::getProductId)
                .collect(Collectors.toSet());

        // 获取相似用户的行为数据
        List<UserBehaviorEntity> similarUserBehaviors = new ArrayList<>();
        for (Map.Entry<Long, Double> entry : userSimilarities.entrySet()) {
            Long similarUserId = entry.getKey();
            similarUserBehaviors.addAll(userBehaviorRepository.queryByUserId(similarUserId));
        }

        // 计算商品评分
        JavaPairRDD<Long, Float> itemScores = javaSparkContext.parallelize(similarUserBehaviors)
            .mapToPair(behavior -> {
                float weight = behavior.getUserBehaviorScore();
                return new Tuple2<>(behavior.getProductId(), weight);
            })
            .reduceByKey(Float::sum);

        // 排除目标用户已经交互过的商品
        itemScores = itemScores.filter(pair -> !targetUserItems.contains(pair._1()));

        // 按评分排序并取前 topN 个商品
        List<Tuple2<Long, Float>> topItems = itemScores.top(topN, (SerializableComparator<Tuple2<Long, Float>>) (o1, o2) -> o2._2().compareTo(o1._2()));

        // 获取推荐商品
        List<Long> products = new LinkedList<>();
        for (Tuple2<Long, Float> item : topItems) {
            products.add(item._1());
        }

        return productRepository.queryByIds(products);
    }
}